/*
 * Copyright (C) 2017 Pingan, Inc. All Rights Reserved.
 */

package com.pingan.haofang.searchcloud.index.service.impl;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import com.alibaba.fastjson.JSON;
import com.pingan.haofang.searchcloud.common.dto.PageDTO;
import com.pingan.haofang.searchcloud.common.utils.DateUtils;
import com.pingan.haofang.searchcloud.index.constants.IndexBuildStatus;
import com.pingan.haofang.searchcloud.index.constants.IndexBuildType;
import com.pingan.haofang.searchcloud.index.dao.IndexBuildProgressDao;
import com.pingan.haofang.searchcloud.index.domain.IndexBuildProgress;
import com.pingan.haofang.searchcloud.index.dto.IndexBuildProgressCountDTO;
import com.pingan.haofang.searchcloud.index.dto.IndexBuildProgressDTO;
import com.pingan.haofang.searchcloud.index.dto.IndexBuildProgressQuery;
import com.pingan.haofang.searchcloud.index.dto.IndexBuildProgressTopDTO;
import com.pingan.haofang.searchcloud.index.dto.IndexBuildRunningProgressDTO;
import com.pingan.haofang.searchcloud.index.service.IndexBuildProgressService;

@Service
public class IndexBuildProgressServiceImpl implements IndexBuildProgressService {

    static Logger LOG = LoggerFactory.getLogger(IndexBuildProgressServiceImpl.class);

    private static final String INDEX_BUILD_PROGRESS_KEY = "searchcloud_index_build_running_progress";

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private IndexBuildProgressDao indexBuildProgressDao;

    @Override
    public void updateRunningProgress(IndexBuildRunningProgressDTO dto) {
        String content = JSON.toJSONString(dto);
        redisTemplate.opsForHash().put(INDEX_BUILD_PROGRESS_KEY, String.valueOf(dto.getId()), content);
    }

    @Override
    public PageDTO<IndexBuildProgressDTO> query(IndexBuildProgressQuery query) {
        Page<IndexBuildProgress> daoResult = indexBuildProgressDao.query(query.getProjectCode(), query.getIndexName(),
                query.getIndexBuildType(), query.getStatus(), query.getBeginDate(), query.getEndDate(),
                new PageRequest(query.getPageNo() - 1, query.getPageSize()));

        PageDTO<IndexBuildProgressDTO> result = PageDTO.of(daoResult, item -> {
            IndexBuildProgressDTO dto = new IndexBuildProgressDTO();
            BeanUtils.copyProperties(item, dto);

            // 如果是运行中状态，则去REDIS中实时获取运行过程信息
            if (dto.getStatus() == IndexBuildStatus.RUNNING.value) {
                IndexBuildRunningProgressDTO progressInfo = getRunningInfo(item.getId());
                if(progressInfo != null) {
                    dto.setFetchTotalCount(progressInfo.getFetchTotalCount());
                    dto.setFetchSuccessCount(progressInfo.getFetchSuccessCount());
                    dto.setImportTotalCount(progressInfo.getImportTotalCount());
                    dto.setImportSuccessCount(progressInfo.getImportSuccessCount());
                }
            }
            return dto;
        });

        return result;
    }

    @Override
    public IndexBuildProgressDTO query(long id) {
        IndexBuildProgress progress = indexBuildProgressDao.findOne(id);
        IndexBuildProgressDTO dto = new IndexBuildProgressDTO();
        BeanUtils.copyProperties(progress, dto);

        // 如果是运行中状态，则去REDIS中实时获取运行过程信息
        if (dto.getStatus() == IndexBuildStatus.RUNNING.value) {
            IndexBuildRunningProgressDTO progressInfo = getRunningInfo(id);
            if (progressInfo != null) {
                dto.setFetchTotalCount(progressInfo.getFetchTotalCount());
                dto.setFetchSuccessCount(progressInfo.getFetchSuccessCount());
                dto.setImportTotalCount(progressInfo.getImportTotalCount());
                dto.setImportSuccessCount(progressInfo.getImportSuccessCount());
            }
        }
        return dto;
    }

    @Override
    public Map<String, IndexBuildProgressDTO> query(String projectCode, String indexName, IndexBuildType indexBuildType,
            IndexBuildStatus status) {
        List<IndexBuildProgress> progressList =
                indexBuildProgressDao.query(projectCode, indexName, indexBuildType.value, status.value);

        return progressList.stream().collect(
                Collectors.toMap(IndexBuildProgress::getDatasourceName,
                        new Function<IndexBuildProgress, IndexBuildProgressDTO>() {

                    @Override
                    public IndexBuildProgressDTO apply(IndexBuildProgress t) {
                        IndexBuildProgressDTO dto = new IndexBuildProgressDTO();
                        BeanUtils.copyProperties(t, dto);
                        return dto;
                    }
                }));
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = RuntimeException.class)
    public void updateProgress(IndexBuildProgressDTO dto) {
        IndexBuildProgress progress = new IndexBuildProgress();
        BeanUtils.copyProperties(dto, progress);
        indexBuildProgressDao.save(progress);

        dto.setId(progress.getId());

        if (dto.getStatus() == IndexBuildStatus.SUCCESS.value || dto.getStatus() == IndexBuildStatus.FAIL.value) {
            redisTemplate.opsForHash().delete(INDEX_BUILD_PROGRESS_KEY, String.valueOf(dto.getId()));
        }
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = RuntimeException.class)
    public void stopProgress(String projectCode, String indexName, String datasourceName,
            IndexBuildType indexBuildType) {

        LOG.info("stop inc progress, projectCode:{}, indexName:{}, dataSourceName:{}, type:{}", projectCode, indexName,
                datasourceName, indexBuildType.desc);

        List<IndexBuildProgress> progressList = null;
        if (datasourceName == null) {
            progressList = indexBuildProgressDao.query(projectCode, indexName, indexBuildType.value,
                    IndexBuildStatus.RUNNING.value);
        } else {
            IndexBuildProgress progress =
                    indexBuildProgressDao.get(projectCode, indexName, datasourceName, indexBuildType.value);
            if (progress != null) {
                progressList = Arrays.asList(progress);
            }
        }

        if (CollectionUtils.isNotEmpty(progressList)) {
            for (IndexBuildProgress progress : progressList) {
                progress.setStatus(IndexBuildStatus.STOP.value);
                progress.setExecuteEndTime(new Date());
                progress.setExecuteCostTime(DateUtils.getBetween(progress.getExecuteStartTime(),
                        progress.getExecuteEndTime(), DateUtils.SECOND_RETURN));

                IndexBuildRunningProgressDTO runningInfo = getRunningInfo(progress.getId());
                if (runningInfo != null) {
                    progress.setFetchTotalCount(runningInfo.getFetchTotalCount());
                    progress.setFetchSuccessCount(runningInfo.getFetchSuccessCount());
                    progress.setImportTotalCount(runningInfo.getImportTotalCount());
                    progress.setImportSuccessCount(runningInfo.getImportSuccessCount());
                }

                indexBuildProgressDao.save(progress);
                redisTemplate.opsForHash().delete(INDEX_BUILD_PROGRESS_KEY, String.valueOf(progress.getId()));
            }
        }
    }

    @Override
    public IndexBuildProgressDTO get(String projectCode, String indexName, String datasourceName,
            IndexBuildType indexBuildType) {
        IndexBuildProgress progress =
                indexBuildProgressDao.get(projectCode, indexName, datasourceName, indexBuildType.value);
        if (progress == null) {
            return null;
        }
        IndexBuildProgressDTO dto = new IndexBuildProgressDTO();
        BeanUtils.copyProperties(progress, dto);
        return dto;
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = RuntimeException.class)
    public void flushRunningProgress() {
        Set<Object> progressIdSet = redisTemplate.opsForHash().keys(INDEX_BUILD_PROGRESS_KEY);
        for (Object key : progressIdSet) {
            Long id = Long.parseLong((String) key);
            IndexBuildRunningProgressDTO runningInfo = getRunningInfo(id);

            IndexBuildProgress progress = indexBuildProgressDao.findOne(id);

            if (runningInfo != null && progress != null) {
                progress.setFetchTotalCount(runningInfo.getFetchTotalCount());
                progress.setFetchSuccessCount(runningInfo.getFetchSuccessCount());
                progress.setImportTotalCount(runningInfo.getImportTotalCount());
                progress.setImportSuccessCount(runningInfo.getImportSuccessCount());
                indexBuildProgressDao.save(progress);
            }
        }
    }

    /**
     * 从REDIS中获取运行时状态信息
     * 
     * @param id
     * @return
     */
    @Override
    public IndexBuildRunningProgressDTO getRunningInfo(long id) {
        String content = (String) redisTemplate.opsForHash().get(INDEX_BUILD_PROGRESS_KEY, String.valueOf(id));

        IndexBuildRunningProgressDTO dto = null;
        if (StringUtils.isNotBlank(content)) {
            try {
                dto = JSON.parseObject(content, IndexBuildRunningProgressDTO.class);
            } catch (Exception e) {
                LOG.error("parse running progress content failed, please check, id:" + id, e);
            }
        }
        return dto;
    }

    @Override
    public List<IndexBuildProgressDTO> queryByStatusAndUpdateTime(Date startTime, Date endTime) {
        List<IndexBuildProgress> data = indexBuildProgressDao.queryByStatusAndUpdateTime(startTime, endTime,
                Arrays.asList(IndexBuildStatus.SUCCESS.value, IndexBuildStatus.FAIL.value));
        List<IndexBuildProgressDTO> result = new ArrayList<>();
        for (IndexBuildProgress progress : data) {
            IndexBuildProgressDTO dto = new IndexBuildProgressDTO();
            BeanUtils.copyProperties(progress, dto);
            result.add(dto);
        }
        return result;
    }

    @Override
    public List<IndexBuildProgressTopDTO> queryIndexBuildProgressCountList(Date fromDate, Date toDate) {
        List<IndexBuildProgressCountDTO> dtoList = indexBuildProgressDao.queryIndexBuildProgressCountList(
                fromDate, toDate, Arrays.asList(IndexBuildStatus.FAIL.getValue(), IndexBuildStatus.SUCCESS.getValue()));


        Map<String, IndexBuildProgressTopDTO> map = new LinkedHashMap<>();

        for (IndexBuildProgressCountDTO item : dtoList) {
            String projectCode = item.getProjectCode();
            String indexName = item.getIndexName();
            String key = projectCode + "-" + indexName;

            IndexBuildProgressTopDTO topDTO = map.get(key);

            if (topDTO == null) {
                topDTO = new IndexBuildProgressTopDTO();
                topDTO.setProjectCode(item.getProjectCode());
                topDTO.setIndexName(item.getIndexName());
            }

            Long execCount = item.getExecCount() == null ? 0 : item.getExecCount();

            if (item.getStatus() == IndexBuildStatus.SUCCESS.getValue()) {
                //调用成功量
                topDTO.setSuccessNum(execCount);
                //索引成功总量
                Long indexTotal = item.getIndexTotal() == null ? 0 : item.getIndexTotal();
                topDTO.setIndexTotal(indexTotal);
                //耗时总量
                Long costTimeTotal = item.getCostTimeTotal() == null ? 0 : item.getCostTimeTotal();
                topDTO.setCostTimeTotal(costTimeTotal);
            }

            if (item.getStatus() == IndexBuildStatus.FAIL.getValue()) {
                //调用失败量
                topDTO.setFailNum(execCount);
                //耗时总量
                Long costTimeTotal = item.getCostTimeTotal() == null ? 0 : item.getCostTimeTotal();
                topDTO.setCostTimeTotal(costTimeTotal);
            }

            map.put(item.getProjectCode() + "-" + item.getIndexName(), topDTO);
        }

        List<IndexBuildProgressTopDTO> topDTOS = map.entrySet().stream().map(x -> {
            IndexBuildProgressTopDTO topDTO = x.getValue();
            if (topDTO == null) {
                return null;
            }

            Long failNum = topDTO.getFailNum() == null ? 0 : topDTO.getFailNum();
            Long successNum = topDTO.getSuccessNum() == null ? 0 : topDTO.getSuccessNum();

            Long totalNum = failNum + successNum;
            topDTO.setExecTotalNum(totalNum);

            Double failRate = Double.parseDouble(failNum + "") / totalNum;
            double rate = new BigDecimal(Double.valueOf(failRate * 100)).
                    setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
            topDTO.setFailRate(rate);
            return topDTO;
        }).collect(Collectors.toList());

        return topDTOS;
    }
}
